package com.wjk.kylin.rpc.core.invoker.future;

import com.wjk.kylin.rpc.core.invoker.RpcConsumerFactory;
import com.wjk.kylin.rpc.core.netty.params.RpcRequest;
import com.wjk.kylin.rpc.core.netty.params.RpcResponse;
import com.wjk.kylin.rpc.core.util.RpcException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * call back future
 *
 * @author wjk
 */
public class RpcFutureResponse {
    private static final Logger LOGGER = LoggerFactory.getLogger(RpcFutureResponse.class);

    private final RpcConsumerFactory consumerFactory;

    private final Thread thread;
    /**
     * net data
     */
    private final RpcRequest request;
    /**
     * volatile
     */
    private RpcResponse response;
    /**
     * future volatile
     */
    private boolean done = false;
    private final Object lock = new Object();

    public RpcFutureResponse(RpcConsumerFactory consumerFactory, RpcRequest request) {
        this.consumerFactory = consumerFactory;
        this.request = request;
        this.thread = Thread.currentThread();

        putConsumerFuture();
    }

    public void putConsumerFuture() {
        this.consumerFactory.putConsumerFuture(request.getRequestId(), this);
    }

    public void removeConsumerFuture() {
        this.consumerFactory.removeConsumerFuture(request.getRequestId());
    }

    //LockSupport 造成 java.lang.Thread.State: WAITING (parking)
    //会报timeout异常
    //park和unpark都会存在失效的情况，不知道什么原因
    //还是使用synchronized 保险
    //防止线程是因中断操作唤醒
//    if (Thread.currentThread().isInterrupted()){
//        LOGGER.error(">>>>>>>>>>> kylin-rpc request isInterrupted start thread is {},requestId = {}", Thread.currentThread(), request.getRequestId());
//        throw new RpcException("线程被中断，异常结束");
//    }
//
//    public void setResponse(RpcResponse response) {
//        this.response = response;
//        this.done = true;
//        LOGGER.debug(">>>>>>>>>>> kylin-rpc response start unpark thread is {},current thread is {},requestId = {}", thread, Thread.currentThread(), request.getRequestId());
//        //唤醒
//        LockSupport.unpark(thread);
//        LOGGER.debug(">>>>>>>>>>> kylin-rpc response end   unpark thread is {},current thread is {},requestId = {}", thread, Thread.currentThread(), request.getRequestId());
//    }
//
//    public RpcResponse get(long timeout, TimeUnit unit) {
//        LOGGER.debug(">>>>>>>>>>> kylin-rpc request start park thread is {},requestId = {}", Thread.currentThread(), request.getRequestId());
//        //阻塞线程
//        if (timeout < 0) {
//            LockSupport.park();
//        } else {
//            long timeoutMillis = (TimeUnit.MILLISECONDS == unit) ? timeout : TimeUnit.MILLISECONDS.convert(timeout, unit);
//            LockSupport.parkUntil(System.currentTimeMillis() + timeoutMillis);
//        }
//        LOGGER.debug(">>>>>>>>>>> kylin-rpc request end   park thread is {},requestId = {}", Thread.currentThread(), request.getRequestId());
//        if (!done) {
//            LOGGER.error(">>>>>>>>>>> kylin-rpc request timeout={}, done = {},requestId = {},response = {}", timeout, done, request.getRequestId(), response);
//            throw new RpcException("kylin-rpc request timeout at:" + System.currentTimeMillis() + ", request:" + request.toString());
//        }
//        return response;
//    }

    public void setResponse(RpcResponse response) {
        this.response = response;
        LOGGER.debug(">>>>>>>>>>> kylin-rpc response start synchronized thread is {},, current thread is {},requestId = {}", thread, Thread.currentThread(), request.getRequestId());
        synchronized (lock) {
            LOGGER.debug(">>>>>>>>>>> kylin-rpc response start done thread is {},, current thread is {},requestId = {}", thread, Thread.currentThread(), request.getRequestId());
            done = true;
            LOGGER.debug(">>>>>>>>>>> kylin-rpc response start notifyAll thread is {},, current thread is {},requestId = {}", thread, Thread.currentThread(), request.getRequestId());
            lock.notifyAll();
        }
    }

    public RpcResponse get() throws InterruptedException {
        return get(-1, TimeUnit.MILLISECONDS);
    }

    public RpcResponse get(long timeout, TimeUnit unit) throws InterruptedException {
        if (!done) {
            LOGGER.debug(">>>>>>>>>>> kylin-rpc request start synchronized thread is {},requestId = {}", Thread.currentThread(), request.getRequestId());
            synchronized (lock) {
                LOGGER.debug(">>>>>>>>>>> kylin-rpc request start done thread is {},requestId = {}", Thread.currentThread(), request.getRequestId());
                if (!done) {
                    LOGGER.debug(">>>>>>>>>>> kylin-rpc request start wait thread is {},requestId = {}", Thread.currentThread(), request.getRequestId());
                    try {
                        if (timeout < 0) {
                            lock.wait();
                        } else {
                            long timeoutMillis = (TimeUnit.MILLISECONDS == unit) ? timeout : TimeUnit.MILLISECONDS.convert(timeout, unit);
                            lock.wait(timeoutMillis);
                        }
                    } catch (InterruptedException e) {
                        throw e;
                    }
                }
            }
        }
        if (!done) {
            throw new RpcException("kylin-rpc request timeout at:" + System.currentTimeMillis() + ", request:" + request.toString());
        }
        return response;
    }


}
